package test

import org.apache.spark.sql.SparkSession

import java.util.regex.Pattern
object gg {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext

  //观影数据分析
    val filePath = "src/main/resources/"
    val  moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath +"ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")

    occupationsRDD.cache()
    usersRDD.cache()
    ratingsRDD.cache()
    moviesRDD.cache()

    //读取数据，统计条数
    println("电影数：" + moviesRDD.count())
    println("用户数：" + usersRDD.count())
    println("职业数：" + occupationsRDD.count())
    println("评分数：" + ratingsRDD.count())

  //根据评级文件获取用户看过的电影数量和对应电影信息
   // val userWatched = ratingsRDD.map(_.split("::"))
     // .map(user => (user(1),user(0)))
      //.filter(_._2.equals("1"))
     // println("id为1用户观看的电影数：" + userWatched.count())
  //观看的电影详情
    //val moviesInfo = moviesRDD.map(_.split("::"))
    //  .map(movie => {(movie(0),(movie(1),movie(2)))
     // })
    //userWatched.join(moviesInfo)
      //.map(item => {
       //  (item._1,item._2._1,item._2._2._1,item._2._2._2)
     // }).foreach(println)


    //练习

    val rating = ratingsRDD.map(x => x.split("::")).map {
     x => {
            (x(0), x(1), x(2)) // (UserID,MovieID,Rating)
        }
    }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
     //  ( MovieId,( Rating, 1) )  (1200,(4.0,1))
    rating.map(x => (x._2, (x._3.toDouble, 1)))
    //  ( MovieId,( 总分, 总次数) )
    .reduceByKey((x, y) => {
      (x._1 + y._1, x._2 + y._2)   })
    // (  平均分, MovieId )
    .map(x => (x._2._1 / x._2._2, x._1))
      .sortByKey(false)
      .take(5)
      .foreach(println)
    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
    // MovieID::Title::Genres
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      //  ( MovieId,( Rating, 1) )  (1252,(4.0,1))
       .reduceByKey((x, y) => {  (x._1 + y._1, x._2 + y._2)
    }) //  ( MovieId,( 总分, 总次数) )
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2))) //( MovieId, (平均分, 总分,总次数) )

    moviesInfo.join(ratingsInfo)
    .map(info => {
      (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
       // (平均分,(movieId,Title,Genres,总分,总次数))
      }).sortByKey(false)   .take(5)   .foreach(println)
      println("观影人数最多的前 10 部电影")
       // (UserID,MovieID,Rating)
    val watchViewsInfo = rating.map(x => {
      (x._2, 1)
    }).reduceByKey((x, y) => x + y) //  ( MovieId,总次数 )
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5) // 5 名
     watchViewsInfo.foreach(println(_))
    println("===================>")
    rating.map(x => (x._2, 1)) // ( MovieId, 1)
     .reduceByKey((x, y) => {   (x + y)   }) //  ( MovieId,总次数 )
     .map(x => (x._2, x._1)) //   (  总次数, MovieId )
      .sortByKey(false)
      .take(10)
      .foreach(println) //  286-> 999
    println("详情的输出(  观影人数，电影编号)")
    // 输出 ( 总次数, (MovieID, title,Genres,总分,平均分  ) )
    //  ratingsInfo   ( MovieId, (平均分, 总分,总次数) )
    //  ( MovieID,  (  (Title,Genres),(平均分, 总分,总次数)  ) )
    // ratingsInfo ( MovieId, (平均分, 总分,总次数) )
    // moviesInfo  MovieID::Title::Genres
    moviesInfo.join(ratingsInfo).map(x => {
        (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
       }).sortByKey(false)
         .take(10)
         .foreach(println)

    val movieStyle = moviesRDD.map(_.split("::"))
     .map(tp => (tp(0),tp(2))) //只留编号和类型
      .flatMapValues(sp => sp.split("\\|")) //切分电影类型
      .map(p => (p._2,1)) //转换成键值对
      .reduceByKey((x,y) => x + y) //聚合
      .filter(_._1.equals("Comedy"))
      movieStyle.foreach(println)
    // movieStyle.take(5).foreach(println)
    //  打印喜剧片（Comedy）的数量


    // "movies.dat"：MovieID::Title::Genres
    // 7::Sabrina (1995)::Comedy|Romance
    // Toy Story (1995) (.*) (\\(\\d{4}\\))
    val pattern = Pattern.compile(" (.*) (\\(\\d{4}\\))")
    val movieYear = moviesRDD.map(_.split("::")).map(x =>  //(MovieID,Title)
         (x(1), 1))
    .map(x => {
      //var name = ""
      var year = ""
      val matcher = pattern.matcher(x._1)
      if (matcher.find()) {
        year = matcher.group(2)
        year = year.substring(1, year.length() - 1)
      }
      year
    })
      .filter(_=="2000").count()
    println(s"2000年产生的电影数：$movieYear")


    sc.stop()
  }
}
